package com.jxstjh.conch.handler.receiver;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.jxstjh.conch.common.domain.AnchorInfo;
import com.jxstjh.conch.common.domain.LogParam;
import com.jxstjh.conch.common.domain.TaskInfo;
import com.jxstjh.conch.common.enums.AnchorStateEnum;
import com.jxstjh.conch.handler.pending.Task;
import com.jxstjh.conch.handler.pending.TaskPendingHolder;
import com.jxstjh.conch.handler.utils.GroupIdMappingUtil;
import com.jxstjh.conch.support.utils.LogUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Scope;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

/**
 * 消费Kafka里所有的消息
 */
@Slf4j
@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class Receiver {

    private static final String LOG_BIZ_TYPE = "Receiver#consumer";

    @Autowired
    private ApplicationContext context;

    @Autowired
    private TaskPendingHolder taskPendingHolder;

    @Autowired
    private LogUtil logUtil;

    /**
     * 无论消息中心接入多少应用, 入口的topic只有一个, 配置在 conch.business.topic.name
     * KafkaListener 没有配置groupId是因为针对不同的发送渠道和消息类型, groupId是不一样的
     * 我希望的是多group消费同一个topic, 所以groupId由ReceiverStart的groupIdEnhancer动态生成, 具体见
     * https://docs.spring.io/spring-kafka/docs/2.7.11/reference/html/#kafkalistener-attrs
     *
     * @param consumerRecord
     * @param topicGroupId
     */
    @KafkaListener(topics = "#{'${conch.business.topic.name}'}")
    public void consumer(ConsumerRecord<?, String> consumerRecord, @Header(KafkaHeaders.GROUP_ID) String topicGroupId) {
        Optional<String> kafkaMessage = Optional.ofNullable(consumerRecord.value());
        if (kafkaMessage.isPresent()) {

            List<TaskInfo> taskInfoLists = JSON.parseArray(kafkaMessage.get(), TaskInfo.class);

            String messageGroupId = GroupIdMappingUtil.getGroupIdByTaskInfo(CollUtil.getFirst(taskInfoLists.iterator()));

            /**
             * 每个消费者组 只消费 他们自身的消息
             */
            if (topicGroupId.equals(messageGroupId)) {
                for (TaskInfo taskInfo : taskInfoLists) {
                    logUtil.print(LogParam.builder().bizType(LOG_BIZ_TYPE).object(taskInfo).build(), AnchorInfo.builder().ids(taskInfo.getReceiver()).businessId(taskInfo.getBusinessId()).state(AnchorStateEnum.RECEIVE.getCode()).build());
                    Task task = context.getBean(Task.class).setTaskInfo(taskInfo);
                    taskPendingHolder.route(topicGroupId).execute(task);
                }
            }
        }
    }
}